package cn.z2huo.demo.kafka.spring.retry.nonblock.consumer;

import cn.hutool.core.util.RandomUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * <p>
 *
 * @author z2huo
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class Cluster1KafkaConsumer {

    @KafkaListener(
            id = "tempTopicKafkaListener",
            containerFactory = "cluster1KafkaListenerContainerFactory",
            topics = "${spring.kafka.cluster1-topic.temp}"
    )
    public void tempListen(@Payload String message) {
        log.info("temp topic receive message data is {}", message);
    }

    @KafkaListener(
            id = "retryTestTopicKafkaListener",
            containerFactory = "cluster1KafkaListenerContainerFactory",
            topics = "${spring.kafka.cluster1-topic.retry-test}"
    )
    public void listen(@Payload String message) {
        boolean exceptionFlag = RandomUtil.randomBoolean();
        if (exceptionFlag) {
            log.info("error, retry topic receive message data is {}", message);
            throw new RuntimeException("error, retry topic receive message data");
        }
        log.info("retry topic receive message data is {}", message);
    }

}
